package com.gaofei.mail.kafka;

import com.gaofei.mail.utils.MsgUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * 收消息的类
 *
 */
@Component
public class KafkaMsg {

    /*@Autowired
    private RedisTemplate<String ,String> redisTemplate;*/

    @Autowired
    MsgUtil msgUtil;

    @KafkaListener(topics = {"yuekao"},containerFactory="ackContainerFactory" ,id="ack")
    public void msg(ConsumerRecord<String, String> msgObject, Acknowledgment ack){
        String brokerKey = msgObject.key();
        String msg = msgObject.value();
        if(msg.equals("pass")){
            //发送邮件给用户提醒
            msgUtil.sendTextEmail("79527743@qq.com","贷款审批通知!","您的贷款申请通过!");
        }
        //setNx  set if not exist
        //如果返回true,就代表着,首次存,能存进去  false:第二次存,存不进去了
//        Boolean b = redisTemplate.opsForValue().setIfAbsent(brokerKey, "sss");
//        if(!b){
//            System.out.println("重复消费");//错误的思路
//        }else{
////            继续消费
//            System.out.println("消息是:"+msgObject.value());//已经收到消息了
//            //开始修改数据的代码
//            System.out.println("修改数据库");
//        }
        //手工提交,让broker移动offsets-->让指针偏移,这行代码一定要放在消费者代码执行完之后
        ack.acknowledge();

    }
}
